package gossip.v2;

import gossip.v2.cluster.ClusterManager;
import gossip.v2.core.StateManager;
import gossip.v2.core.TopologyManager;
import gossip.v2.core.TransportManager;
import gossip.v2.message.SyncMessage;
import gossip.v2.message.UserMessage;
import gossip.v2.cluster.GossipNode;
import lombok.Getter;
import lombok.Setter;
import org.ricks.ioc.utils.Logger;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

// Gossip核心组件
@Getter
public class GossipCore {
    @Getter
    private final GossipNode localNode;
    @Getter
    private final ClusterManager clusterManager;
    private final StateManager stateManager;
    @Getter
    private final TransportManager transport;
    private final TopologyManager topologyManager;
    @Getter
    private final ScheduledExecutorService scheduler;
    @Getter
    @Setter
    private List<GossipListener> listeners = new CopyOnWriteArrayList<>();
    private final GossipServer server;
    private final AtomicBoolean stopping = new AtomicBoolean(false);
    @Setter
    private int gossipInterval = 1000;
    @Setter
    private int fanout = 3;
    @Setter
    private int fullSyncThreshold = 50;
    @Setter private int suspicionTimeout = 5000;  // 可配置参数
    @Setter private int removalTimeout = 10000;

    public GossipCore(GossipNode localNode) {
        this.localNode = localNode;
        this.localNode.setLocal(true);
        this.clusterManager = new ClusterManager(localNode);
        this.stateManager = new StateManager(localNode);
        this.transport = new TransportManager(this);
        this.topologyManager = new TopologyManager();

        this.scheduler = Executors.newScheduledThreadPool(2, r -> {
            Thread t = new Thread(r, "gossip-worker");
            t.setDaemon(true);
            return t;
        });

        this.clusterManager.setCore(this);
        this.stateManager.setCore(this);

        this.server = new GossipServer(localNode.getPort(), this);

        // 连接事件通知
        this.clusterManager.addListener(new ClusterManager.NodeStateListener() {
            @Override
            public void onNodeJoined(GossipNode node) {
                // 触发主动连接
                if (!node.equals(localNode) && !node.isLocal()) {
                    transport.connectToNode(node);
                }
                listeners.forEach(l -> l.onNodeJoined(node));
            }

            @Override
            public void onNodeLeft(GossipNode node) {
                // 清理重连上下文
                transport.onReconnectFailure(node.getNodeId());
                listeners.forEach(l -> l.onNodeLeft(node));
            }
        });

        // 传递超时配置
        this.clusterManager.setSuspicionTimeout(suspicionTimeout);
        this.clusterManager.setRemovalTimeout(removalTimeout);
    }

    public void start() {
        server.start();
        clusterManager.initSeedConnections();

        // 优化：更紧凑的初始执行
        scheduler.scheduleWithFixedDelay(this::doGossipCycle, 0, gossipInterval, TimeUnit.MILLISECONDS);
        scheduler.scheduleWithFixedDelay(clusterManager::detectFailedNodes, 1, 3, TimeUnit.SECONDS);
        // 添加定时健康检查
        scheduler.scheduleWithFixedDelay(() -> {
            if (isStopping()) return;

            // 检查所有已知节点连接
            clusterManager.getAllKnownNodes().stream()
                    .filter(id -> !id.equals(localNode.getNodeId()))
                    .forEach(id -> {
                        if (!transport.isConnectionHealthy(id)) {
                            GossipNode node = clusterManager.getNodeById(id);
                            if (node != null) {
                                transport.scheduleReconnect(node);
                            }
                        }
                    });
        }, 30, 30, TimeUnit.SECONDS);
    }


    private void doGossipCycle() {
        try {
            List<GossipNode> targets = clusterManager.selectRandomNodes(fanout);
            for (GossipNode target : targets) {

                // 使用新增的 getStateDifferenceSize 方法
                int diffSize = stateManager.getStateDifferenceSize(
                        target.getVersionVector());

                if (diffSize > fullSyncThreshold) {
                    sendFullSync(target);
                } else if (diffSize > 0) {  // 只有当有差异时才发送增量同步
                    sendSyncMessage(target);
                }
            }
        } catch (Exception e) {
            Logger.error("Gossip cycle error", e);
        }
    }

    private void sendSyncMessage(GossipNode target) {
        StateManager.DeltaState delta = stateManager.getDeltaState(
                target.getVersionVector());

        if (!delta.isEmpty()) {
            transport.send(target, new SyncMessage(
                    localNode,
                    delta.getVersions(),
                    delta.getStateValues(),
                    false
            ));
        }
    }


    private void sendFullSync(GossipNode target) {
        transport.send(target, new SyncMessage(
                localNode,
                stateManager.getVersions(),
                stateManager.getState(),
                true
        ));
    }


    public void updateState(String key, byte[] value) {
        stateManager.update(key, value);
        transport.broadcast(new UserMessage(localNode, key, value));
    }

    public void stop() {
        if (!stopping.compareAndSet(false, true)) return;

        // 1. 关闭所有会话
        transport.shutdown();

        // 2. 停止服务器
        server.shutdown();

        // 3. 优雅关闭调度器
        scheduler.shutdown();
        try {
            if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) {
                scheduler.shutdownNow();
            }
        } catch (InterruptedException e) {
            scheduler.shutdownNow();
        }

        // 4. 清理资源
        clusterManager.getLiveNodes().clear();
        clusterManager.getSuspiciousNodes().clear();
    }

    public boolean isStopping() {
        return stopping.get();
    }

    public interface GossipListener {
        void onNodeJoined(GossipNode node);
        void onNodeLeft(GossipNode node);
        void onStateChanged(String key, Object value);
    }
}
